package com.five.consumer;

import com.five.esearch.INovelDataSyncService;
import com.five.mapper.NovelMapper;
import com.five.mq.NovelSyncMessage;
import com.five.pojo.Novel;
import com.five.util.NovelIndexInitializer;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 小说数据同步消息消费者
 */
@Component
@RocketMQMessageListener(
        topic = "novel-topic",  // 监听的主题
        consumerGroup = "reader-service-consumer-group"  // 消费者组
)
public class NovelSyncConsumer implements RocketMQListener<NovelSyncMessage> {

    @Autowired
    private NovelMapper novelMapper;
    @Autowired
    private INovelDataSyncService dataSyncService;
    @Autowired
    private RestHighLevelClient esClient;
    @Autowired
    private NovelIndexInitializer indexInitializer;  // 索引工具类

    @Override
    public void onMessage(NovelSyncMessage message) {
        try {
            Long novelId = message.getNovelId();
            String operation = message.getOperation();

            switch (operation) {
                case "insert":
                case "update":
                    // 查询MySQL最新数据
                    Novel novel = novelMapper.selectById(novelId);
                    if (novel != null) {
                        // 同步到ES（复用已有转换方法）
                        dataSyncService.syncSingleNovel(novel);
                    }
                    break;
                case "delete":
                    // 从ES删除对应文档
                    DeleteRequest deleteRequest = new DeleteRequest(
                            indexInitializer.INDEX_NAME,
                            indexInitializer.TYPE_NAME,
                            novelId.toString()
                    );
                    esClient.delete(deleteRequest, RequestOptions.DEFAULT);
                    break;
                default:
                    throw new IllegalArgumentException("未知操作类型：" + operation);
            }
        } catch (Exception e) {
            // 消息消费失败，可记录日志并触发重试（RocketMQ默认支持重试）
            throw new RuntimeException("ES同步失败：" + e.getMessage(), e);
        }
    }
}